package com.yanxu;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2025-05-19-2025
 * @Description: flink 批处理
 */
public class BatchWordCount {


    /*
    *  使用 flink 进行批处理。
    *  批处理一般用在有着 【明确的数据边界】 的情况下。
    *  比如在这个例子中， 数据源就是一个固定的  world.txt 文本， 这里有着明确的数据边界。
    * */
    public static void main(String[] args) throws Exception {
        // 创建执行环境，这是Flink程序的基础，它代表了程序将要执行的环境。
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 数据源
        // 读取文本文件作为数据源，这里读取的是"input/words.txt"中的内容。
        DataSource<String> lineDS = env.readTextFile("flink-01/src/main/resources/world.txt");
        // 对读取的每一行数据应用flatMap操作，将每行文本分割成单词，并为每个单词生成一个Tuple2<String, Long>对，其中String是单词，Long初始化为1。
        // 这里Lambda表达式用于定义如何处理输入的每一行数据，并输出到Collector。
        FlatMapOperator<String, Tuple2<String, Long>> wordAndOne =
                // 相当于每次处理 一行文本
                // lineDS.flatMap((String line, Collector<Tuple2<String, Long>> out)
                /*
                lineDS.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
                    private static final long serialVersionUID = 1L;
                    @Override
                    public void flatMap(String s, Collector<Tuple2<String, Long>> collector) throws Exception {

                    }
                });*/

                // 使用更加简便的 lambda 表达式的方式写法
                // String line： 输入参数
                // Collector<Tuple2<String, Long>> out：输出参数
                lineDS.flatMap((String line, Collector<Tuple2<String, Long>> out) -> {
                            // 使用空格分隔每一行文本，得到单词数组
                            String[] words = line.split(" ");
                            // 遍历分割后的所有单词
                            for (String word : words) {
                                /**
                                 * 字段0 字段1
                                 * Hello 1
                                 * World 1
                                 * Hello 1
                                 * Lily 1
                                 */
                                out.collect(Tuple2.of(word, 1L)); // 为每个单词生成一个元组(word, 1)，并收集到结果中
                            }
                        })
                        .returns(Types.TUPLE(Types.STRING, Types.LONG));

        // wordAndOne.groupBy(0),  groupBy 分组函数。
        // 根据Tuple2中的第一个元素（即单词）进行无序分组，准备进行聚合操作。
        UnsortedGrouping<Tuple2<String, Long>> wordAndOneUG = wordAndOne.groupBy(0);

        /**
         * 字段0 字段1
         * Hello 2
         * World 1
         * Lily 1
         */
        // wordAndOneUG.sum(1); 分完组之后，根据第二列(索引是1)。 进行求和操作。
        // 对分组后的数据按第二个元素（计数值）进行求和，得到每个单词的总出现次数。
        AggregateOperator<Tuple2<String, Long>> sum = wordAndOneUG.sum(1);

        // 打印最终结果，即每个单词及其出现的总次数。
        sum.print();
    }
}